package com.spider.moudle.proxy;

/**
 * @author 夏茂轩
 * @package com.spider.silence.proxy
 * @Date 2017/7/28 13:50
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.google.common.collect.Sets;
import com.spider.moudle.constant.Constant;
import com.spider.moudle.entity.Proxy;
import com.spider.moudle.factory.HttpClientFactory;
import org.apache.http.HttpHost;
import org.apache.http.client.HttpRequestRetryHandler;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.protocol.HttpContext;
import org.eclipse.jetty.util.ConcurrentHashSet;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class HttpProxyContext {

    private Set<Proxy> orgProxy =new ConcurrentHashSet<>();
    /**
     * 存储每个downloader的代理
     */
    private Map<Integer, LinkedBlockingDeque<Proxy>> storage = Maps.newHashMap();

    private ExecutorService proxyCheckPool = Executors.newFixedThreadPool(500);

    private LinkedBlockingDeque<Proxy> copy = Queues.newLinkedBlockingDeque();

    private static HttpProxyContext instance;

    private HttpProxyContext() {

    }

    public static synchronized HttpProxyContext getInstance() {    //对获取实例的方法进行同步
        if (instance == null) {
            instance = new HttpProxyContext();
            instance.init();
        }
        return instance;
    }

    private String get(String url) throws IOException {
        String content = null;
        CloseableHttpClient httpClient = null;
        HttpHost p = null;
        BufferedReader reader = null;
        CloseableHttpResponse res = null;
        HttpGet get = null;
        HttpRequestRetryHandler my = new HttpRequestRetryHandler() {
            @Override
            public boolean retryRequest(
                    IOException exception,
                    int executionCount,
                    HttpContext context) {
                return false;
            }
        };
        try {
            httpClient = HttpClientFactory.buildHttpClient();
            get = new HttpGet(url);
            get.addHeader("Accept", "Accept text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8");
            get.addHeader("User-Agent", "Mozilla/5.0 (Windows NT 6.1; rv:6.0.2) Gecko/20100101 Firefox/6.0.2");
            get.addHeader("Connection", "close");
            res = httpClient.execute(get);
            if (res.getStatusLine().getStatusCode() == 200) {
                reader = new BufferedReader(new InputStreamReader(res.getEntity().getContent()));
                content = reader.readLine();
            } else {
            }

        } finally {
            if (reader != null) {
                reader.close();
            }
            if (get != null) {
                get.releaseConnection();
            }
            if (res != null) {
                res.close();
            }
        }
        return content;
    }

    private void init() {
        //System.out.println("预热代理池");
        List<Proxy> proxies = loadNewProxy();
        proxies = checkProxy(proxies);
        synchronized (orgProxy) {
            List<Proxy> xr = new ArrayList<>();
            orgProxy.addAll(proxies);
            xr.addAll(orgProxy);
            xr = xr.stream().filter(s->{
                if(s.getResponse_lsp()>=Constant.PROXY_STANDARD){
                    return false;
                }else {
                    return true;
                }
            }).sorted((s1, s2) -> {
                if (s1.getResponse_lsp() > s2.getResponse_lsp()) {
                    return 1;
                } else {
                    return -1;
                }
            }).limit(150).collect(Collectors.toCollection(ArrayList::new));
            //限制数量最高100个
            orgProxy.clear();
            orgProxy.addAll(xr);
            //System.out.println("代理池预热完毕，高质量代理IP有" + orgProxy.size() + "个" + "-------" + String.format("%tF %tT", new Date(), new Date()));
            orgProxy.notifyAll();
        }
        //线程做循环检查IP的质量
        Thread thread = new Thread(new OrgProxyChecker());
        thread.start();
        //线程做循环load最新的高质量的代理IP进来，并淘汰现有IP中质量比较低的IP，并更新downloader的代理队列
        Thread thread2 = new Thread(new ProxyLoader(), "代理加载器");
        thread2.start();
    }

    private List<Proxy> loadNewProxy() {
        List<Proxy> proxies = Lists.newArrayList();
        String rts = "";
        while (true) {
            try {
                rts = get(Constant.HTTP_PROXY_GET_URL);
                JSONArray array = JSON.parseObject(rts).getJSONObject("data").getJSONArray("proxy_list");
                for (int i = 0; i < array.size(); i++) {
                    String[] proxy = array.getString(i).split(":");
                    proxies.add(new Proxy(proxy[0], Integer.parseInt(proxy[1])));
                }
                break;
            } catch (Exception e) {
                try {
                    Thread.sleep(1000 * 2);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                //e.printStackTrace();
            }
        }
        return proxies;
    }

    private List<Proxy> checkProxy(List<Proxy> proxyList) {
        List<Proxy> proxies = Lists.newArrayList();
        List<FutureTask<Proxy>> futureTasks = new ArrayList<FutureTask<Proxy>>();
        for (int i = 0; i < proxyList.size(); i++) {
            ProxyChecker checker = new ProxyChecker(proxyList.get(i));
            FutureTask<Proxy> futureTask = new FutureTask<Proxy>(checker);
            futureTasks.add(futureTask);
            proxyCheckPool.submit(futureTask);
        }
        for (int i = 0; i < futureTasks.size(); i++) {
            try {
                Proxy proxy = futureTasks.get(i).get(Constant.PROXY_STANDARD + 500, TimeUnit.MILLISECONDS);
                if (proxy != null && proxy.getResponse_lsp() < Constant.PROXY_STANDARD + 500) {
                    proxies.add(proxy);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
            }
        }
        return proxies;
    }

    private class ProxyChecker implements Callable<Proxy> {
        Proxy proxy = null;

        public ProxyChecker(Proxy proxy) {
            this.proxy = proxy;
        }

        @Override
        public Proxy call() throws Exception {
            long start = System.currentTimeMillis();
            try {
                get("http://www.baidu.com/", proxy);
                long end = System.currentTimeMillis();
                if (proxy.getResponse_lsp() != 0) {
                    double lsp = end - start;
                    proxy.setResponse_lsp((lsp + proxy.getResponse_lsp()) / 2d);
                } else {
                    double lsp = end - start;
                    proxy.setResponse_lsp(lsp);
                }
                //System.out.println("高质量代理：{proxy:{IP:'" + proxy.getIp() + "',PORT:" + proxy.getPort() + "},response_time：" + (proxy.getResponse_lsp()) + "}" + "-------" + String.format("%tF %tT", new Date(), new Date()));
            } catch (Exception e) {
                if (proxy != null) {
                    if (proxy.getResponse_lsp() != 0) {
                        proxy.setResponse_lsp((Constant.PROXY_STANDARD + 1000 + proxy.getResponse_lsp()) / 2d);
                        //System.out.println("代理[" + proxy.getIp() + ":" + proxy.getPort() + "]......返回时间：" + (proxy.getResponse_lsp()) + "ms");
                    } else {
                        proxy.setResponse_lsp(Constant.PROXY_STANDARD + proxy.getResponse_lsp() + 3000);
                    }
                }
            }
            return proxy;
        }

        private void get(String url, Proxy proxy) throws Exception {
            HttpRequestRetryHandler my = new HttpRequestRetryHandler() {
                public boolean retryRequest(
                        IOException exception,
                        int executionCount,
                        HttpContext context) {
                    return false;
                }
            };
            CloseableHttpClient httpClient = null;
            HttpHost p = null;
            CloseableHttpResponse res = null;
            HttpGet get = null;
            try {
                p = new HttpHost(proxy.getIp(), proxy.getPort());
                httpClient = HttpClientFactory.buildHttpClient(p);
                get = new HttpGet(url);
                get.addHeader("Accept", "Accept text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8");
                get.addHeader("User-Agent", "Mozilla/5.0 (Windows NT 6.1; rv:6.0.2) Gecko/20100101 Firefox/6.0.2");
                get.addHeader("Connection", "close");
                res = httpClient.execute(get);
                if (res.getStatusLine().getStatusCode() == 200) {
                    return;
                } else {
                    throw new Exception("超时");
                }
            } finally {
                if (res != null) {
                    res.close();
                }
                if (get != null) {
                    get.releaseConnection();
                }
            }
        }
    }

    private class OrgProxyChecker implements Runnable {
        Proxy[] array = {};

        @Override
        public void run() {
            while (true) {
                try {
                    Thread.sleep(Constant.PROXY_LOOP_CHECK_TIME_INTEVAL);
                    synchronized (orgProxy) {
                        Proxy[] array = orgProxy.toArray(new Proxy[orgProxy.size()]);
                        this.array = Arrays.copyOf(array, array.length);
                        orgProxy.notifyAll();
                    }
                    List<Proxy> proxies = Arrays.asList(array);
                    proxies = checkProxy(proxies);
                    synchronized (orgProxy) {
                        List<Proxy> xr = new ArrayList<>();
                        orgProxy.addAll(proxies);
                        xr.addAll(orgProxy);
                        xr = xr.stream().filter(s->{
                            if(s.getResponse_lsp()>=Constant.PROXY_STANDARD){
                                return false;
                            }else {
                                return true;
                            }
                        }).sorted((s1, s2) -> {
                            if (s1.getResponse_lsp() > s2.getResponse_lsp()) {
                                return 1;
                            } else {
                                return -1;
                            }
                        }).limit(150).collect(Collectors.toCollection(ArrayList::new));
                        //限制数量最高100个
                        orgProxy.clear();
                        orgProxy.addAll(xr);
                        orgProxy.notifyAll();
                    }
                    for (Map.Entry<Integer, LinkedBlockingDeque<Proxy>> entry : storage.entrySet()) {
                        synchronized (storage) {
                            storage.put(entry.getKey(), getCopyOfHighQualityProxy());
                            storage.notifyAll();
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private class ProxyLoader implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    Thread.sleep(Constant.PROXY_LOOP_LOAD_TIME_INTEVAL);
                    List<Proxy> proxies = loadNewProxy();
                    proxies = checkProxy(proxies);
                    synchronized (orgProxy) {
                        orgProxy.addAll(proxies);
                        List<Proxy> listX = orgProxy.stream().filter(s1->{
                            if(s1.getResponse_lsp()>=Constant.PROXY_STANDARD){
                                return false;
                            }else {
                                return true;
                            }
                        }).sorted((s1, s2) -> {
                            if (s1.getResponse_lsp() > s2.getResponse_lsp()) {
                                return 1;
                            } else {
                                return -1;
                            }
                        }).limit(150).collect(Collectors.toCollection(ArrayList::new));
                        orgProxy.clear();
                        orgProxy.addAll(listX);
                        System.err.println("当前高质量IP池大小:" + orgProxy.size() + "-------" + String.format("%tF %tT", new Date(), new Date()));
                        orgProxy.removeAll(orgProxy.parallelStream().filter(proxy -> proxy.getResponse_lsp() > (Constant.PROXY_STANDARD + 500)).collect(Collectors.toCollection(ArrayList::new)));
                        orgProxy.notifyAll();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 获取高质量的代理的复制品
     *
     * @return
     */
    private LinkedBlockingDeque<Proxy> getCopyOfHighQualityProxy() {
        LinkedBlockingDeque<Proxy> proxies = new LinkedBlockingDeque<>();
        synchronized (orgProxy) {
            proxies = orgProxy.parallelStream().map(proxy -> {
                return new Proxy(proxy.getIp(), proxy.getPort(), proxy.getResponse_lsp());
            }).sorted(new Comparator<Proxy>() {
                @Override
                public int compare(Proxy o1, Proxy o2) {
                    int code = 0;
                    if (o1.getResponse_lsp() > o2.getResponse_lsp()) {
                        code = 1;
                    } else {
                        code = -1;
                    }
                    return code;
                }
            }).collect(Collectors.toCollection(LinkedBlockingDeque::new));
            orgProxy.notifyAll();
        }
        return proxies;
    }

    public Proxy fetchProxy() {
        while (true) {
            if (copy.size() == 0) {
                synchronized (copy) {
                    copy.addAll(getCopyOfHighQualityProxy());
                    copy.notifyAll();
                }
            }
            Proxy proxy = copy.poll();
            if (proxy == null) {
                continue;
            } else {
                return proxy;
            }
        }
    }

    /**
     * 获取当前downLoader的代理池
     *
     * @return
     */
    public ArrayList<Proxy> getProxyPoolOfDownLoader() {
        ArrayList<Proxy> proxies = orgProxy.stream().sorted(new Comparator<Proxy>() {
            @Override
            public int compare(Proxy o1, Proxy o2) {
                int code = 0;
                if (o1.getResponse_lsp() > o2.getResponse_lsp()) {
                    code = 1;
                } else {
                    code = -1;
                }
                return code;
            }
        }).collect(Collectors.toCollection(ArrayList::new));
        return proxies;
    }
}